package com.ruoyi.iot.listener;


import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.ruoyi.iot.service.IDeviceDataService;
import com.ruoyi.iotcommon.constant.RocketTopicConstant;
import com.ruoyi.iotcommon.domain.DeviceDataMessage;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 设备上报数据监听器，可以看一下6-02之前的提交记录，是来一条数据插入一次数据库，不是批量插入
 * 另一种批量消费方案：https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-4-x-series/developer-reference/batch-consumption
 */
@Slf4j
@Service
@EnableScheduling
@RocketMQMessageListener( topic = RocketTopicConstant.DEVICE_DATA, consumerGroup = "iot-core-device-data")
public class DeviceDataListener implements RocketMQListener<DeviceDataMessage>{

 
    @Resource
    IDeviceDataService deviceDataService;

    // 添加批量写入队列
    private final BlockingQueue<DeviceDataMessage> writeQueue = new LinkedBlockingQueue<>(10000);

    // 每秒批量插入一次数据库，提升性能
    @Scheduled(fixedRate = 1000)
    public void reportCurrentTime() {
        if (writeQueue.isEmpty()) {
            return;
        }
        List<DeviceDataMessage> batchDeviceData = new ArrayList<>();
        writeQueue.drainTo(batchDeviceData);
        deviceDataService.insertDevicePropertyData(batchDeviceData);
    }


    @Override
    public void onMessage(DeviceDataMessage message) {
        log.info("iot-core 收到设备上报数据:{}", JSON.toJSONString(message));
        boolean success = writeQueue.offer(message);
        if (!success) {
            log.warn("写入队列失败，队列已满");
            // 可以选择记录到本地文件或发送告警
        }
    }
    
}
